package top.jolyoulu.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @Author: JolyouLu
 * @Date: 2024/5/19 15:07
 * @Description
 */
object Spark01_SparkStreaming_WordCount {
  def main(args: Array[String]): Unit = {
    //创建环境对象
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    //参数1：环境配置
    //参数2：批处理的周期，采集的周期
    val ssc = new StreamingContext(sparkConf,Seconds(3))

    //处理逻辑
    //获取端口数据
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

    val words: DStream[String] = lines.flatMap(_.split(" "))

    val wordToOne: DStream[(String, Int)] = words.map((_, 1))

    val wordToCount: DStream[(String, Int)] = wordToOne.reduceByKey(_ + _)

    wordToCount.print()

    //启动采集器
    ssc.start()
    //等待采集器的关闭
    ssc.awaitTermination()
  }
}
